package me.itsoo.artemis.framework.aop.util;

import me.itsoo.artemis.framework.core.thread.ThreadFactoryBuilder;

import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * 延迟任务缓冲执行，避免瞬时的重复并发（局部削峰）
 *
 * <ol>
 *   <li>每100ms刷新当前延迟任务列表</li>
 *   <li>初始化延迟100ms执行</li>
 * </ol>
 *
 * @author zxy
 * @version 0.0.1
 * @since 2023/6/2
 */
public class DelayedTaskBuffer<T> {

    private final AtomicInteger pointer;

    private final ConcurrentSkipListSet<T>[] timerTask;

    private final ScheduledExecutorService timerExecutor;

    private final Consumer<T> taskConsumer;

    @SuppressWarnings("unchecked")
    public DelayedTaskBuffer(Consumer<T> taskConsumer) {
        this.pointer = new AtomicInteger(-1);
        this.timerTask = new ConcurrentSkipListSet[10];
        this.timerExecutor = new ScheduledThreadPoolExecutor(1, getDefaultThreadFactory());
        this.taskConsumer = taskConsumer;

        initial();
    }

    private ThreadFactory getDefaultThreadFactory() {
        return new ThreadFactoryBuilder().setNameFormat("delay-timer-task-%d").build();
    }

    /**
     * 放入延迟执行任务
     *
     * @param t 延迟任务参数
     */
    public void push(T t) {
        final int curr = (pointer.get() + 1) % timerTask.length;
        timerTask[curr].add(t);
    }

    /**
     * 初始化延迟任务容器（线程安全）
     */
    private void initial() {
        for (int j = 0; j < timerTask.length; j++) {
            timerTask[j] = new ConcurrentSkipListSet<>();
        }

        startListening();
    }

    /**
     * 开始任务监听
     */
    private void startListening() {
        timerExecutor.scheduleWithFixedDelay(() -> {
            increment();
            // 消费任务
            final Iterator<T> it = all().iterator();
            while (it.hasNext()) {
                taskConsumer.accept(it.next());
                it.remove();
            }
        }, 100, 100, TimeUnit.MILLISECONDS);
    }

    /**
     * 指针自增
     */
    private void increment() {
        int curr;
        int next;

        do {
            curr = pointer.get();
            next = (curr + 1) % timerTask.length;
        } while (!pointer.compareAndSet(curr, next));
    }

    /**
     * 获取全部延迟任务
     *
     * @return Set
     */
    private Set<T> all() {
        final int curr = pointer.get();
        return timerTask[curr];
    }
}
